package cn.doitedu.demo4.calculators;

import cn.doitedu.demo4.beans.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import redis.clients.jedis.Jedis;

import java.io.IOException;

/**
 *    静态画像条件: age > 20
 *    动态画像条件: 规则上线后，A事件要发生3+次
 *    跨区间统计条件： 2023-11-01 00:00:00.000 ~ 未来 ,X事件要发生5次+
 *    触发条件：  W事件
 */
public class Model_1_RuleCalculator implements RuleCalculator{
    Roaring64Bitmap preSelectUserBitmap;
    Long history_end_time;
    String ruleId;
    JSONObject paramObject;
    JSONObject message;

    ValueState<Integer> cntState;
    ValueState<Integer> crossState;

    Jedis jedis;


    @Override
    public void init(RuntimeContext runtimeContext, String ruleParam, Roaring64Bitmap preSelectUserBitmap, Long history_end_time) {

        this.preSelectUserBitmap = preSelectUserBitmap;
        this.history_end_time = history_end_time;

        this.paramObject = JSON.parseObject(ruleParam);
        this.ruleId = paramObject.getString("rule_id");

        message = new JSONObject();
        message.put("rule_id",ruleId);


        // 申请一个用于统计实时画像事件次数的状态
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-realtime", Integer.class));

        // 申请一个用于统计跨区间条件目标事件次数的状态
        crossState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-cross", Integer.class));


        // 构建redis连接客户端
        jedis = new Jedis("doitedu", 6379);

    }

    @Override
    public void calculate(EventBean eventBean, Collector<String> out) throws IOException {

        Long userId = eventBean.getUser_id();

        // 判断该事件的行为人，是否是本规则的目标人群
        if(!preSelectUserBitmap.contains(userId)) return;

        // 因为本规则，是有跨区间条件的，所以，在flink的主代码中，会对该运算机第一次运行时回放最近10s的数据
        // 为了防止重复计算，需要判断回放的数据，是否早于  历史值统计截止时间点
        if(eventBean.getAction_time() < history_end_time)  return;


        // 取实时画像条件各类参数
        JSONObject realtimeProfile = paramObject.getJSONObject("realtime_profile");
        String realtimeProfileEventId = realtimeProfile.getString("event_id");
        int realtimeProfileMinCount = realtimeProfile.getIntValue("min_count");

        // 取跨区间条件各类参数
        JSONObject crossRange = paramObject.getJSONObject("cross_range");
        Long crossRangeStart = crossRange.getLong("start_time");
        Long crossRangeEnd = crossRange.getLong("end_time");
        String crossRangeEventId = crossRange.getString("event_id");
        int crossRangeMinCount = crossRange.getIntValue("min_count");
        String conditionId = crossRange.getString("condition_id");

        // 取触发条件参数
        String triggerEvent = paramObject.getString("trigger_event");

        String eventId = eventBean.getEvent_id();
        /*  1,动态画像统计逻辑处理 */
        // 如果当前事件，是动态画像的目标事件
        if(eventId.equals(realtimeProfileEventId)){
            // 计数+1
            cntState.update(cntState.value()==null? 1 : cntState.value()+1);
        }


        /*  2,区间条件统计逻辑处理 */
        if(eventId.equals(crossRangeEventId)){
            // 计数+1
            // 判断计数state中是否为空
            if(crossState.value() == null){
                // 说明该用户的历史值还没有取过，就需要取redis中取
                //  redis中的数据结构： Hash
                //      大key  ==> rule_id
                //         小key  ==>  user_id:condition_id
                //         值     ==> 历史值
                String redisHistoryValueStr = jedis.hget(ruleId, userId + ":" + conditionId);
                int redisHistoryValueInt = redisHistoryValueStr == null? 0 : Integer.parseInt(redisHistoryValueStr);
                crossState.update( redisHistoryValueInt+ 1);
            }else{
                crossState.update( crossState.value() + 1);
            }
        }


        /*  3,触发条件判断逻辑 */
        if(eventId.equals(triggerEvent)){


            int crossConditionValue =0;
            if(crossState.value()==null){
                String redisHistoryValueStr = jedis.hget(ruleId, userId + ":" + conditionId);
                crossConditionValue = redisHistoryValueStr == null? 0 : Integer.parseInt(redisHistoryValueStr);
            }else{
                crossConditionValue = crossState.value();
            }


            // 判断该用户的动态画像条件是否满足
            // 判断该用户的跨区间统计条件是否满足
            if(cntState.value()!=null
                    &&  cntState.value()> realtimeProfileMinCount
                    &&  crossConditionValue > crossRangeMinCount){
                // 如果都满足，则输出命中消息
                message.put("hit_time",eventBean.getAction_time());
                message.put("user_id",userId);

                out.collect(message.toJSONString());
            }
        }

    }
}
